package spark_study

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

object coregroup {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setAppName("coregroup").setMaster("local[*]")
    val sc = new SparkContext(conf)
    //广播变量
    val rdd5: RDD[(String, String)] = sc.parallelize(List(("tom", "汤姆"), ("jerry", "杰瑞"), ("jack", "杰克")))
    val rdd1: RDD[(String, String)] = sc.parallelize(List(("tom", "1,2,3"), ("jerry", "3,4,5"), ("tom", "5,6,7")))
    val rdd2: RDD[(String, String)] = sc.parallelize(List(("tom", "1,2,3"), ("jack", "3,4,5"), ("rose", "5,6,7")))
    val rdd3: RDD[(String, (Iterable[String], Iterable[String]))] = rdd1.cogroup(rdd2)
    //收集到driver端
    val arr: Array[(String, String)] = rdd5.collect()
    val broadInDriver: Broadcast[Array[(String, String)]] = sc.broadcast(arr)
    //得到(名字,数量)
    val rdd4: RDD[(String, Int)] = rdd3.mapValues((v: (Iterable[String], Iterable[String])) => {
      //v,_1
      val iter1: Iterable[List[Int]] = v._1.map(_.split(",").map(_.toInt).toList)
      val iter2: Iterable[List[Int]] = v._2.map(_.split(",").map(_.toInt).toList)
      val sum1: Int = iter1.map(_.sum).sum
      val sum2: Int = iter2.map(_.sum).sum
      sum1+sum2
    })
    //定义可变数组
    //取出rdd4中元素与广播变量匹配

//    rdd4.map(e=>{
//      for (elem <- br) yield {
//        val arr1 = new ArrayBuffer[(String, Int)]()
//        if (elem._1.equals(e._1)){
//          arr1.append((elem._2, e._2))
//        }
//      }
//    })

      val rdd6: RDD[(String, Int)] = rdd4.map(e => {
        val br: Array[(String, String)] = broadInDriver.value
        var index =0
        var index1=0
        var flag=false
        println(br.length)
        while(index < br.length){
          if (br(index)._1 == e._1)
          {
            index1=index
            index+=1
            flag=true
          } else {
            index+=1
           }
         }
        if (flag){

          (br(index1)._2,e._2)
        }else{
          (e._1,e._2)
        }
      })
    println(rdd6.collect().toBuffer)
    sc.stop()
  }
}
